package com.ferry.transport.netty.channel;

import com.ferry.transport.api.channel.FutureListener;
import com.ferry.transport.api.channel.WrappedChannel;
import com.ferry.transport.netty.consumer.ConnectionKeeper;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoop;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import io.netty.util.internal.PlatformDependent;

import java.net.SocketAddress;
import java.util.Queue;

/**
 * Created by daiyong
 */
public class NettyChannel implements WrappedChannel {


	private static final AttributeKey<NettyChannel> NETTY_CHANNEL_KEY = AttributeKey.valueOf("netty.channel");

	private final Channel channel;

	private final Queue<Runnable> taskQueue = PlatformDependent.newMpscQueue(1024);

	private final Runnable runAllTasks = this::runAllTasks;

	public static NettyChannel attachChannel(Channel channel) {
		Attribute<NettyChannel> attr = channel.attr(NETTY_CHANNEL_KEY);
		NettyChannel nChannel = attr.get();
		if (nChannel == null) {
			NettyChannel newNChannel = new NettyChannel(channel);
			nChannel = attr.setIfAbsent(newNChannel);
			if (nChannel == null) {
				nChannel = newNChannel;
			}
		}
		return nChannel;
	}


	public NettyChannel(Channel channel) {
		this.channel = channel;
	}

	public Channel channel() {
		return channel;
	}

	@Override
	public String id() {
		return channel.id().asShortText(); // 注意这里的id并不是全局唯一, 单节点中是唯一的
	}

	@Override
	public boolean isActive() {
		return channel.isActive();
	}

	@Override
	public boolean inIoThread() {
		return channel.eventLoop().inEventLoop();
	}

	@Override
	public SocketAddress localAddress() {
		return channel.localAddress();
	}

	@Override
	public SocketAddress remoteAddress() {
		return channel.remoteAddress();
	}

	@Override
	public boolean isWritable() {
		return channel.isWritable();
	}

	@Override
	public boolean isMarkedReconnect() {
		ConnectionKeeper connectionKeeper = channel.pipeline().get(ConnectionKeeper.class);
		return connectionKeeper != null && connectionKeeper.isStarted();
	}

	@Override
	public boolean isAutoRead() {
		return channel.config().isAutoRead();
	}

	@Override
	public void setAutoRead(boolean autoRead) {
		channel.config().setAutoRead(autoRead);
	}

	@Override
	public WrappedChannel close() {
		channel.close();
		return this;
	}

	@Override
	public WrappedChannel close(final FutureListener<WrappedChannel> listener) {
		final WrappedChannel channel = this;
		this.channel.close().addListener((ChannelFutureListener) future -> {
			if (future.isSuccess()) {
				listener.success(channel);
			} else {
				listener.failure(channel, future.cause());
			}
		});
		return channel;
	}

	@Override
	public WrappedChannel write(Object msg) {
		channel.writeAndFlush(msg, channel.voidPromise());
		return this;
	}

	@Override
	public WrappedChannel write(Object msg, final FutureListener<WrappedChannel> listener) {
		final WrappedChannel channel = this;
		this.channel.writeAndFlush(msg)
				.addListener((ChannelFutureListener) future -> {
					if (future.isSuccess()) {
						listener.success(channel);
					} else {
						listener.failure(channel, future.cause());
					}
				});
		return channel;
	}

	@Override
	public void addTask(Runnable task) {
		EventLoop eventLoop = channel.eventLoop();

		while (!taskQueue.offer(task)) {
			if (eventLoop.inEventLoop()) {
				runAllTasks.run();
			} else {
				eventLoop.execute(runAllTasks);
			}
		}

		if (!taskQueue.isEmpty()) {
			eventLoop.execute(runAllTasks);
		}
	}

	private void runAllTasks() {
		if (taskQueue.isEmpty()) {
			return;
		}

		for (;;) {
			Runnable task = taskQueue.poll();
			if (task == null) {
				return;
			}
			task.run();
		}
	}

	@Override
	public boolean equals(Object obj) {
		return this == obj || (obj instanceof NettyChannel && channel.equals(((NettyChannel) obj).channel));
	}

	@Override
	public int hashCode() {
		return channel.hashCode();
	}

	@Override
	public String toString() {
		return channel.toString();
	}

}
